﻿using Apex.EventBus.Connections;
using Apex.EventBus.Handlers;
using Microsoft.Extensions.DependencyInjection;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System.Diagnostics;
using System.Reflection;
using System.Text;
using System.Text.Json;

namespace Apex.EventBus.EventBuses
{
    class RabbitMQEventBus : IEventBus, IDisposable
    {
        private IModel _consumerChannel;
        private readonly string _exchangeName;
        private string _queueName;
        private readonly RabbitMQConnection _persistentConnection;
        private readonly SubscriptionsManager _subsManager;
        private readonly IServiceProvider _serviceProvider;
        private readonly IServiceScope serviceScope;
        public RabbitMQEventBus(RabbitMQConnection persistentConnection,IServiceScopeFactory serviceProviderFactory, string exchangeName, string queueName)
        {
            this._persistentConnection = persistentConnection ?? throw new ArgumentNullException(nameof(persistentConnection));
            this._subsManager = new SubscriptionsManager();
            this._exchangeName = exchangeName;
            this._queueName = queueName;

            // 创建一个服务作用域以解析作用域服务
            this.serviceScope = serviceProviderFactory.CreateScope();
            this._serviceProvider = serviceScope.ServiceProvider;
            this._consumerChannel = CreateConsumerChannel();
            this._subsManager.OnEventRemoved += SubsManager_OnEventRemoved; ;
        }
        public void Dispose()
        {
            // 释放消费者通道，清除订阅，释放持久连接和服务作用域
            if (_consumerChannel != null)
            {
                _consumerChannel.Dispose();
            }
            _subsManager.Clear();
            this._persistentConnection.Dispose();
            this.serviceScope.Dispose();
        }

        public void Publish(string eventName, object? eventData)
        {
            // 如果持久连接未连接，则连接
            if (!_persistentConnection.IsConnected)
            {
                _persistentConnection.TryConnect();
            }

            // 创建一个用于发布消息的通道
            using (var channel = _persistentConnection.CreateModel())
            {
                // 声明交换机
                channel.ExchangeDeclare(exchange: _exchangeName, type: "direct");

                // 将事件数据序列化为 JSON
                byte[] body;
                if (eventData == null)
                {
                    body = new byte[0];
                }
                else
                {
                    JsonSerializerOptions options = new JsonSerializerOptions
                    {
                        WriteIndented = true
                    };
                    body = JsonSerializer.SerializeToUtf8Bytes(eventData, eventData.GetType(), options);
                }

                // 设置消息属性
                var properties = channel.CreateBasicProperties();
                properties.DeliveryMode = 2; // 持久化

                // 将消息发布到交换机
                channel.BasicPublish(
                    exchange: _exchangeName,
                    routingKey: eventName,
                    mandatory: true,
                    basicProperties: properties,
                    body: body);
            }
        }

        public void Subscribe(string eventName, Type handlerType)
        {
            CheckHandlerType(handlerType);
            DoInternalSubscription(eventName);
            _subsManager.AddSubscription(eventName, handlerType);
            StartBasicConsume();
        }

        public void Unsubscribe(string eventName, Type handlerType)
        {
            CheckHandlerType(handlerType);
            _subsManager.RemoveSubscription(eventName, handlerType);
        }

        /// <summary>
        /// SubscriptionsManager 中的 OnEventRemoved 事件的事件处理程序
        /// </summary>
        /// <param name="sender">事件发送者</param>
        /// <param name="eventName">事件名称</param>
        private void SubsManager_OnEventRemoved(object? sender, string eventName)
        {
            // 如果持久连接未连接，则重新连接
            if (!_persistentConnection.IsConnected)
            {
                _persistentConnection.TryConnect();
            }

            // 取消绑定已删除事件的队列和交换机
            using (var channel = _persistentConnection.CreateModel())
            {
                channel.QueueUnbind(queue: _queueName,
                    exchange: _exchangeName,
                    routingKey: eventName);

                // 如果没有订阅事件，则关闭消费者通道
                if (_subsManager.IsEmpty)
                {
                    _queueName = string.Empty;
                    _consumerChannel.Close();
                }
            }
        }

        /// <summary>
        /// 创建用于消费消息的通道
        /// </summary>
        /// <returns>消费者通道。</returns>
        private IModel CreateConsumerChannel()
        {
            // 如果持久连接未连接，则重新连接
            if (!_persistentConnection.IsConnected)
            {
                _persistentConnection.TryConnect();
            }

            // 创建用于消费消息的通道
            var channel = _persistentConnection.CreateModel();
            channel.ExchangeDeclare(exchange: _exchangeName, type: "direct");

            channel.QueueDeclare(queue: _queueName,
                                 durable: true,
                                 exclusive: false,
                                 autoDelete: false,
                                 arguments: null);

            // 处理通道的回调异常
            channel.CallbackException += (sender, ea) =>
            {
                Debug.Fail(ea.ToString());
            };

            return channel;
        }

        /// <summary>
        /// 检查处理程序类型是否实现了 IIntegrationEventHandler 接口
        /// </summary>
        /// <param name="handlerType">处理程序类型。</param>
        /// <exception cref="ArgumentException">处理程序类型未实现 IIntegrationEventHandler 接口时抛出。</exception>
        private void CheckHandlerType(Type handlerType)
        {
            // 检查处理程序类型是否实现了 IIntegrationEventHandler 接口
            if (!typeof(IIntegrationEventHandler).IsAssignableFrom(handlerType))
            {
                throw new ArgumentException($"{handlerType} 不继承自 IIntegrationEventHandler", nameof(handlerType));
            }
        }

        /// <summary>
        /// 执行内部订阅
        /// </summary>
        /// <param name="eventName">事件名称。</param>
        private void DoInternalSubscription(string eventName)
        {
            // 检查事件是否有订阅者
            var containsKey = _subsManager.HasSubscriptionsForEvent(eventName);
            if (!containsKey)
            {
                // 如果持久连接未连接，则重新连接
                if (!_persistentConnection.IsConnected)
                {
                    _persistentConnection.TryConnect();
                }

                // 将队列绑定到交换机以处理该事件
                _consumerChannel.QueueBind(queue: _queueName,
                                    exchange: _exchangeName,
                                    routingKey: eventName);
            }
        }

        /// <summary>
        /// 开始基本的消息消费
        /// </summary>
        private void StartBasicConsume()
        {
            // 检查消费者通道是否已经初始化
            if (_consumerChannel != null)
            {
                // 创建用于接收消息的消费者
                var consumer = new AsyncEventingBasicConsumer(_consumerChannel);
                consumer.Received += Consumer_Received; // 添加消息接收事件处理程序
                _consumerChannel.BasicConsume(
                    queue: _queueName,
                    autoAck: false,
                    consumer: consumer); // 开始基本消费
            }
        }

        /// <summary>
        /// 消息接收事件的处理程序
        /// </summary>
        /// <param name="sender">事件发送者</param>
        /// <param name="eventArgs">事件参数</param>
        private async Task Consumer_Received(object sender, BasicDeliverEventArgs eventArgs)
        {
            var eventName = eventArgs.RoutingKey; // 获取事件名称
            var message = Encoding.UTF8.GetString(eventArgs.Body.Span); // 获取消息内容

            try
            {
                // 处理接收到的事件
                await ProcessEvent(eventName, message);

                // 确认消息
                _consumerChannel.BasicAck(eventArgs.DeliveryTag, multiple: false);
            }
            catch (Exception ex)
            {
                // 处理处理错误
                Debug.Fail(ex.ToString());
            }
        }

        /// <summary>
        /// 处理接收到的事件
        /// </summary>
        /// <param name="eventName">事件名称</param>
        /// <param name="message">消息内容</param>
        private async Task ProcessEvent(string eventName, string message)
        {
            // 检查是否有订阅了指定事件的处理程序
            if (_subsManager.HasSubscriptionsForEvent(eventName))
            {
                var subscriptions = _subsManager.GetHandlersForEvent(eventName); // 获取订阅了该事件的处理程序类型列表
                foreach (var subscription in subscriptions)
                {
                    // 创建解析作用域以解析作用域服务
                    using var scope = this._serviceProvider.CreateScope();
                    // 从作用域中获取对应处理程序类型的服务实例
                    IIntegrationEventHandler? handler = scope.ServiceProvider.GetService(subscription) as IIntegrationEventHandler;
                    if (handler == null)
                    {
                        throw new ApplicationException($"无法创建 {subscription} 类型的服务");
                    }

                    // 处理事件
                    await handler.Handle(eventName, message);
                }
            }
            else
            {
                // 如果找不到处理事件的处理程序，则输出警告信息
                string entryAsm = Assembly.GetEntryAssembly().GetName().Name;
                Debug.WriteLine($"找不到可以处理 eventName={eventName} 的处理程序，entryAsm: {entryAsm}");
            }
        }
    }
}
